/**
 * 
 */
package com.ffcs.crmd.tsp.core.netty.server;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.UnknownHostException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import com.ffcs.crmd.tsp.api.netty.INettyConfigServer;
import com.ffcs.crmd.tsp.api.netty.INettyProcessor;
import com.ffcs.crmd.tsp.api.netty.INettyRemotingServer;
import com.ffcs.crmd.tsp.common.exception.TspRemotingException;
import com.ffcs.crmd.tsp.core.netty.NettyRemotingAbstract;
import com.ffcs.crmd.tsp.core.netty.RpcPair;
import com.ffcs.crmd.tsp.core.thread.ThreadFactoryImpl;
import com.ffcs.crmd.tsp.core.util.RemotingUtil;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.concurrent.DefaultEventExecutorGroup;

/**
 * 功能说明:
 *
 * @author ZHONGFUHUA
 * 
 * @since 
 *
 */
public abstract class NettyRemotingServerAbstract extends NettyRemotingAbstract implements INettyRemotingServer {
    
    /**
     * netty启动NIO服务的辅助类 
     */
    private final ServerBootstrap     serverBootstrap          = new ServerBootstrap();
    
    /**
     * I/O操作的多线程事件循环器
     */
    private final EventLoopGroup      workGroup;
    
    /**
     * I/O操作的多线程事件循环器
     */
    private final EventLoopGroup      bossGroup;
    
    /**
     * 通信服务端处理请求线程池
     */
    private final ExecutorService     processsorExecutor;
    
    /**
     * 通信服务端处理请求应答线程池
     */
    private final ExecutorService     callBackExecutor;
    
    /**
     * 本地server绑定的端口
     */
    private String                    addr;
    
    /**
     * netty客户端配置
     */
    private final INettyConfigServer  nettyConfig;
    
    /**
     * 
     */
    private DefaultEventExecutorGroup defaultGroup;
    
    /**
     * 定时服务
     */
    private ScheduledExecutorService  scheduledExecutorService = Executors
        .newSingleThreadScheduledExecutor(new ThreadFactoryImpl("TspServerNettyRemotingScheduledThread"));
    
    /**
     * 
     * 功能说明:
     * 
     * @param nettyConfig
     */
    public NettyRemotingServerAbstract(INettyConfigServer nettyConfig) {
        super(nettyConfig.getPermitsOneway(), nettyConfig.getPermitsAsync());
        this.nettyConfig = nettyConfig;
        int callBackThreadNums = nettyConfig.getCallbackExecutorThreads();
        int processorThreadNums = nettyConfig.getProcessorExecutorThreads();
        if (callBackThreadNums <= 0) {
            callBackThreadNums = 4;
        }
        if (processorThreadNums <= 0) {
            processorThreadNums = 4;
        }
        
        callBackExecutor = Executors.newFixedThreadPool(callBackThreadNums, new ThreadFactoryImpl("TspNettyServerCallBackExecutor-"));
        processsorExecutor = Executors.newFixedThreadPool(processorThreadNums, new ThreadFactoryImpl("TspNettyServerProcesssorExecutor-"));
        
        bossGroup = new NioEventLoopGroup(1, new ThreadFactoryImpl("TspServerNettyBossLoop-"));
        workGroup = new NioEventLoopGroup(nettyConfig.getSelectorThreads(), new ThreadFactoryImpl("TspServerNettySelectorLoop-"));
        defaultGroup = new DefaultEventExecutorGroup(nettyConfig.getWorkGroupThreads(), new ThreadFactoryImpl("TspServerNettyWorkLoop-"));
    }
    
    @Override
    public void start() throws TspRemotingException {
        try {
            InetAddress inetAddress = InetAddress.getLocalHost();
            serverBootstrap.group(bossGroup, workGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 1024)
                .option(ChannelOption.SO_REUSEADDR, true).option(ChannelOption.SO_KEEPALIVE, false).option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_SNDBUF, nettyConfig.getSocketSndBufSize())
                .option(ChannelOption.SO_RCVBUF, nettyConfig.getSocketRcvBufSize())
                .localAddress(new InetSocketAddress(inetAddress, nettyConfig.getListenPort())).childHandler(new NettyServerInitializer(this));
            
            ChannelFuture cf = serverBootstrap.bind().sync();
            SocketAddress socketAddress = cf.channel().localAddress();
            addr = RemotingUtil.parseSocketAddress(socketAddress);
            if (cf.isSuccess()) {
                getLog().info("SERVER-NETTY: netty remoting server[{}] start ok.", addr);
                System.out.println(String.format("tsp server addr = [%s]", addr));
            }
        } catch (InterruptedException | UnknownHostException e) {
            throw new TspRemotingException("start server netty remoting exception: " + e.getMessage());
        }
        
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    NettyRemotingServerAbstract.this.scanTimeoutAsyncResponseTable();
                } catch (Exception e) {
                    getLog().error("SERVER-NETTY: scan response table exception:", e);
                }
            }
        }, 1000, 60 * 1000, TimeUnit.MILLISECONDS);
    }
    
    @Override
    public void shutdown() {
        workGroup.shutdownGracefully();
        bossGroup.shutdownGracefully();
        
        scheduledExecutorService.shutdown();
        processsorExecutor.shutdown();
        callBackExecutor.shutdown();
        getLog().info("SERVER-NETTY: netty remoting server[{}] shutdown ok.", addr);
    }
    
    @Override
    public INettyRemotingServer registerProcessor(int requestCode, INettyProcessor processor) {
        RpcPair<INettyProcessor, ExecutorService> pair = new RpcPair<INettyProcessor, ExecutorService>(processor, this.processsorExecutor);
        this.getProcessorTable().put(requestCode, pair);
        return this;
    }
    
    @Override
    public int getServerChannelMaxIdleTimeSeconds() {
        return nettyConfig.getServerChannelMaxIdleTimeSeconds();
    }
    
    @Override
    public void removeResponse(int opaque) {
        super.getResponseTable().remove(opaque);
    }
    
    @Override
    public ExecutorService getCallBackExecutor() {
        return callBackExecutor;
    }
    
    @Override
    public ExecutorService getProcesssorExecutor() {
        return processsorExecutor;
    }
    
    /**
     * 功能说明: 获取defaultEventExecutorGroup
     *
     * @return defaultEventExecutorGroup defaultEventExecutorGroup
     */
    @Override
    public DefaultEventExecutorGroup getDefaultEventExecutorGroup() {
        return defaultGroup;
    }
    
    @Override
    public String getAddr() {
        return addr;
    }
    
}
